package com.qing.spark

import java.text.SimpleDateFormat

import com.mongodb.spark.MongoSpark
import com.qing.spark.dao.{MemoryDao}
import org.apache.spark.{SparkConf, SparkContext}
import org.bson.Document

import scala.collection.mutable.ListBuffer
import scala.collection.JavaConversions._

/**
  * Created by wuliao on 2018/4/11.
  */
object DealMongodb {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("ShenGang")
      .setJars(Array("file:///mnt/disk/jar/mongdb/streaming-logger-0.0.1-jar-with-dependencies.jar"))
      .set("spark.mongodb.input.uri", "mongodb://175.102.18.112:27018/shengang.log_data")
      .set("spark.mongodb.output.uri", "mongodb://175.102.18.112:27018/shengang.log_data")
      .setMaster("spark://175.102.18.112:7077")
//      .setMaster("local")

    val sc = new SparkContext(conf)
    val rdd = MongoSpark.load(sc)






    rdd.filter((doc: Document) => {
      doc.getString("counter_name").equals("memory_util")
    })
      .map((doc: Document) => {
        val timestamp = doc.getString("@timestamp")
        val sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'")


        val dt = sdf.parse(timestamp)
        val time = dt.getTime
        val second = (time / 1000).toLong
        val minute = (second / 60).toLong
        val hour = (minute / 60).toLong
        val day = (hour / 24).toInt

        var counter_volume: java.lang.Double = 0.0
        try {
          counter_volume = doc.getDouble("counter_volume")
        } catch {
          case e: Exception => {
            counter_volume = doc.getInteger("counter_volume").toDouble
          }
        }

        val host = doc.getString("host")
        (host + "-" + day, (counter_volume, host, day, 1))
      })
      .reduceByKey(func)
      .map(s => {
        (s._1, s._2._1 / s._2._4, s._2._2, s._2._3)
      })
      .foreachPartition(it => {
        val list = new ListBuffer[Tuple4[java.lang.String, Double, java.lang.String, Int]]
        it.foreach(msg => {
          list.append(msg)
        })
        MemoryDao.save(list)
      })


  }

  def func(pre: Tuple4[java.lang.Double, java.lang.String, Int, Int],
           after: Tuple4[java.lang.Double, java.lang.String, Int, Int]):
  Tuple4[java.lang.Double, java.lang.String, Int, Int] = {
    (pre._1 + after._1, pre._2, pre._3, pre._4 + after._4)
  }


}
